package com.ibm.cps.storm;

import java.util.Map;

import com.ibm.cps.kafka.KafkaTopicConstructorForMultiTenants;
import org.apache.log4j.Logger;

import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

import com.google.common.base.Throwables;
import com.ibm.cps.kafka.KafkaTopicConstructor;
import com.ibm.cps.kafka.KafkaTopicConsumeTemplate;
import com.ibm.cps.message.MessageFactory;
import com.ibm.cps.message.RawDataMessage;
import com.ibm.util.TopologyMessageFieldIds;
import com.ibm.util.TopologyStreamIds;
import com.ibm.util.exception.CPSException;

/**
 * Receive message from kafka message topic
 * @author Jingjing Wang
 *
 * 2015-7-14
 */
@SuppressWarnings("rawtypes")
public class MessageSpout extends BaseRichSpout {
	private static final long serialVersionUID = 1L;
	private Logger logger;
	private KafkaTopicConsumeTemplate messageConsumer;
	private KafkaTopicConstructorForMultiTenants kafkaTopicConstructorForMultiTenants = new KafkaTopicConstructorForMultiTenants();

	@Override
	public void open(Map conf, TopologyContext context,
			final SpoutOutputCollector collector) {
		String zookeeper = (String) conf.get("zookeeper");
		String tenantid = (String) conf.get("tenantid");
		String messageTopic = kafkaTopicConstructorForMultiTenants.getMessageTopic(tenantid);
		logger = Logger.getLogger(MessageSpout.class);

		messageConsumer = new KafkaTopicConsumeTemplate(zookeeper, messageTopic) {

			@Override
			public void process(byte[] message) throws CPSException {
				RawDataMessage msg = (RawDataMessage) MessageFactory
						.produceDataMessage(message);
				if (null != msg) {
					collector.emit(TopologyStreamIds.RAW_DATA_STREAM,
							new Values(msg, msg.getTsKey()));
				}
			}
		};
	}

	public void nextTuple() {
		try {
			messageConsumer.consumeMessage();
		} catch (CPSException e) {
			logger.error(Throwables.getStackTraceAsString(e));
		}
	}

	/**
	 * Ouput <type, message>
	 */
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declareStream(TopologyStreamIds.RAW_DATA_STREAM,
				new Fields(TopologyMessageFieldIds.MESSAGE,
						TopologyMessageFieldIds.TS_KEY));
	}
}
